﻿#include <stdarg.h>
#include <unistd.h>
#include "ngx_global.h"
#include "ngx_func.h"
#include "ngx_c_threadpool.h"
#include "ngx_c_memory.h"
#include "ngx_macro.h"
//静态成员初始化
pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;
bool CThreadPool::m_shutdown = false;
/***************************************************************
 * @file       
 * @brief      构造函数
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
CThreadPool::CThreadPool()
{
    m_iRunningThreadNum = 0;  //正在运行的线程，开始给个0
    m_iLastEmgTime = 0;       //上次报告线程不够用了的时间；
    m_iRecvMsgQueueCount = 0; //收消息队列
}
/***************************************************************
 * @file       
 * @brief      析构函数
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
CThreadPool::~CThreadPool()
{
    //接收消息队列中内容释放
    clearMsgRecvQueue();
}
/***************************************************************
 * @file       
 * @brief      清理接收消息队列
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
void CThreadPool::clearMsgRecvQueue()
{
    char *sTmpMempoint;
    CMemory *p_memory = CMemory::GetInstance();
    while(!m_MsgRecvQueue.empty())
    {
        sTmpMempoint = m_MsgRecvQueue.front();
        m_MsgRecvQueue.pop_front(); 
        p_memory->FreeMemory(sTmpMempoint);
    }
}
/***************************************************************
 * @file       
 * @brief      创建线程
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
bool CThreadPool::Create(int threadNum)
{
    ThreadItem *pNew;
    int err;
    m_iThreadNum = threadNum;
    for(int i = 0; i < m_iThreadNum; ++i)
    {
        /*创建一个新线程对象并入到容器中*/
        m_threadVector.push_back(pNew = new ThreadItem(this));
        /*创建线程，错误不返回到errno，一般返回错误码*/
        err = pthread_create(&pNew->_Handle, NULL, ThreadFunc, pNew);
        if(err != 0)
        {
            ngx_log_stderr(err,"CThreadPool::Create()创建线程%d失败，返回的错误码为%d!",i,err);
            return false;
        }
    }
    std::vector<ThreadItem*>::iterator iter;
lblfor:
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        if( (*iter)->ifrunning == false)
        {
            /*等待所有线程创建完毕100ms*/
            usleep(100 * 1000);
            goto lblfor;
        }
    }
    return true;
}
/***************************************************************
 * @file       
 * @brief      线程入口函数
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
void* CThreadPool::ThreadFunc(void* threadData)
{
    ThreadItem *pThread = static_cast<ThreadItem*>(threadData);
    CThreadPool *pThreadPoolObj = pThread->_pThis;
    CMemory *p_memory = CMemory::GetInstance();
    int err;
    pthread_t tid = pthread_self();
    while(true)
    {
        err = pthread_mutex_lock(&m_pthreadMutex);
        if(err != 0)
        {
            ngx_log_stderr(err,"CThreadPool::ThreadFunc()中\
                pthread_mutex_lock()失败，返回的错误码为%d!",err);
        }
        while((pThreadPoolObj->m_MsgRecvQueue.size() == 0) && m_shutdown == false)
        {
            if(pThread->ifrunning == false)
                pThread->ifrunning = true;
            pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex);
        }

        //先判断线程退出这个条件
        if(m_shutdown)
        {
            pthread_mutex_unlock(&m_pthreadMutex); //解锁互斥量
            break;
        }

        //走到这里，可以取得消息进行处理了消息队列中必然有消息注意，目前还是互斥着呢
        char *jobbuf = pThreadPoolObj->m_MsgRecvQueue.front();     //返回第一个元素但不检查元素存在与否
        pThreadPoolObj->m_MsgRecvQueue.pop_front();                //移除第一个元素但不返回    
        --pThreadPoolObj->m_iRecvMsgQueueCount;                    //收消息队列数字-1

        err = pthread_mutex_unlock(&m_pthreadMutex); 
        if(err != 0)
        {
            ngx_log_stderr(err,"CThreadPool::ThreadFunc()中\
                pthread_mutex_unlock()失败，返回的错误码为%d!",err);
        }

        //能走到这里的，就是有消息可以处理，开始处理
        ++pThreadPoolObj->m_iRunningThreadNum;
        g_socket.threadRecvProcFunc(jobbuf);     //处理消息队列中来的消息
        p_memory->FreeMemory(jobbuf);              //释放消息内存 
        --pThreadPoolObj->m_iRunningThreadNum;     //原子-1记录正在干活的线程数量减少1
    }
    return (void*)0;
}
/***************************************************************
 * @file       
 * @brief      线程回收函数
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
void CThreadPool::StopAll()
{
    if(m_shutdown == true)
    {
        return;
    }
    m_shutdown = true;
    /*唤醒所有线程，这个时候m_shutdown为ture从而退出循环*/
    int err = pthread_cond_broadcast(&m_pthreadCond); 
    if(err != 0)
    {
        ngx_log_stderr(err,"CThreadPool::StopAll()\
            中pthread_cond_broadcast()失败，返回的错误码为%d!",err);
        return;
    }
    std::vector<ThreadItem*>::iterator iter;
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        pthread_join((*iter)->_Handle, NULL);
    }
    pthread_mutex_destroy(&m_pthreadMutex);
    pthread_cond_destroy(&m_pthreadCond);
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        if(*iter)
            delete *iter;
    }
    m_threadVector.clear();
    ngx_log_stderr(0,"CThreadPool::StopAll()成功返回，线程池中线程全部正常结束!");
    return;
}
/***************************************************************
 * @file       
 * @brief      收到一个完整消息后，入消息队列，并触发线程池中线程来处理该消息
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
void CThreadPool::inMsgRecvQueueAndSignal(char *buf)
{
    /*上锁*/
    int err = pthread_mutex_lock(&m_pthreadMutex);
    if(err != 0)
    {
        ngx_log_stderr(err,"CThreadPool::inMsgRecvQueueAndSignal()\
            pthread_mutex_lock()失败，返回的错误码为%d!",err);
    }

    m_MsgRecvQueue.push_back(buf);           //入消息队列
    ++m_iRecvMsgQueueCount;                  //收消息队列数字+1

    /*解锁*/
    err = pthread_mutex_unlock(&m_pthreadMutex);
    if(err != 0)
    {
        ngx_log_stderr(err,"CThreadPool::inMsgRecvQueueAndSignal()\
            pthread_mutex_unlock()失败，返回的错误码为%d!",err);
    }
    /*通知消息队列干活*/
    Call();
    return;
}
/***************************************************************
 * @file       
 * @brief      来任务了，调一个线程池中的线程下来干活
 * @author     txj
 * @version    v1
 * @return    
 * @date       2021/8/12
 **************************************************************/
void CThreadPool::Call()
{
    int err = pthread_cond_signal(&m_pthreadCond);
    if(err != 0 )
    {
        ngx_log_stderr(err,"CThreadPool::Call()中\
            pthread_cond_signal()失败，返回的错误码为%d!",err);
    }
    //如果当前的工作线程全部都忙，则要报警
    if(m_iThreadNum == m_iRunningThreadNum)
    {
        time_t currtime = time(NULL);
        /*每隔10秒报告一次线程不够用*/
        if(currtime - m_iLastEmgTime > 10)
        {
            m_iLastEmgTime = currtime;  //更新时间
            ngx_log_stderr(0,"CThreadPool::Call()中\
                发现线程池中当前空闲线程数量为0，要考虑扩容线程池了!");
        }
    }
    return;
}

//唤醒丢失问题，sem_t sem_write;
//参考信号量解决方案：https://blog.csdn.net/yusiguyuan/article/details/20215591  linux多线程编程--信号量和条件变量 唤醒丢失事件